// Copyright (c) 2020 Presto Labs Pte. Ltd.
// Author: linchuan

#include "coin2/exchange/binance/order_v1/parser.h"

#include <map>
#include <memory>
#include <set>
#include <string>
#include <utility>
#include <vector>

#include "coin/proto/coin_request.pb.h"
#include "coin2/base/conversion.h"
#include "coin2/base/log.h"
#include "coin2/exchange/binance/api_order/native_private_client.h"
#include "coin2/exchange/binance/order_v1/error_code.h"
#include "coin2/exchange/util/json_util.h"

namespace coin2::exchange::binance::order_v1 {

namespace {

using namespace coin2::exchange::binance::order_v1::error_code;

using coin::proto::AccountBalance;
using coin::proto::AccountPosition;
using coin::proto::CancelOrderResponse;
using coin::proto::CurrencyTransferElement;
using coin::proto::FillType;
using coin::proto::ProductFillBundle;
using coin::proto::ProductOrderBundle;
using coin::proto::ProductOrderElement;
using coin::proto::SubmitOrderResponse;
using coin::proto::TransferRequestProto;
using coin2::exchange::base::order::IOrderContext;
using coin2::exchange::binance::symbology::BinanceCurrency;
using coin2::exchange::binance::symbology::BinanceProduct;

using JsonValue = rapidjson::GenericValue<rapidjson::UTF8<>>;
using TransferType = coin::proto::TransferRequestProto::TransferType;
using BinanceOrderSubTopicId = coin2::exchange::binance::api_order::BinanceOrderSubTopicId;

inline int64_t GetTransferTimestamp(std::string time_str) {
  time_str.replace(10, 1, "T");
  return TimestampFromIso8601(time_str + ".000Z");
}

inline int64_t GetProcOrderId(const JsonValue& data) {
  int64_t proc_order_id = 0;
  if (data.HasMember("clientOrderId") && data["clientOrderId"].IsInt64()) {
    proc_order_id = stoll(data["clientOrderId"].GetString());
  }
  return proc_order_id;
}

inline coin::proto::OrderSide GetOrderSide(const std::string& side) {
  if (side == "BUY") {
    return coin::proto::OrderSide::BUY_ORDER;
  } else if (side == "SELL") {
    return coin::proto::OrderSide::SELL_ORDER;
  } else {
    return coin::proto::OrderSide::UNKNOWN_ORDER_SIDE;
  }
}

inline coin::proto::OrderType GetOrderType(const std::string& type) {
  if (type == "LIMIT") {
    return coin::proto::OrderType::LIMIT_ORDER;
  } else if (type == "MARKET") {
    return coin::proto::OrderType::MARKET_ORDER;
  } else {
    return coin::proto::OrderType::UNKNOWN_ORDER_TYPE;
  }
}

void SetOrderStatus(const std::string& status, ProductOrderElement* order) {
  if (status == "PENDING_CANCEL" || status == "REJECTED") {
    order->set_is_live(false);
  } else if (status == "NEW" || status == "CONFIRMED" || status == "PARTIALLY_FILLED") {
    order->set_is_live(true);
  } else if (status == "CANCELED" || status == "EXPIRED") {
    order->set_cancel_confirmed(true);
  } else if (status == "FILLED") {
    order->set_fully_filled(true);
  } else {
    LOG(ERROR) << "[ParseRestOrderInfo] Unknown order status! " << status;
  }
}

bool IsBinanceErrMessage(const JsonValue& doc) {
  return doc.HasMember("code") && doc.HasMember("msg");
}

bool IsSuccessDataRequestResp(const rapidjson::GenericValue<rapidjson::UTF8<>>& json) {
  /*
  {
    "result": [
      "MmONVCQoX4djSGNP5ujxTuGv6xoh21fYio0IGZMxSx6zEwXMbzX30OmlAMJd"
    ],
    "id": "3"
  }
  */
  SPDLOG_INFO("[Binance] recv: {}", JsonToString(json));
  if (!json.HasMember("id") || !json.HasMember("result")) {
    return false;
  }
  if (!json["result"].IsArray()) {
    return false;
  }
  bool ret = false;
  for (const auto& data : json["result"].GetArray()) {
    (void)data;
    ret = true;
    break;
  }
  return ret;
}

bool IsAccountUpdate(const rapidjson::GenericValue<rapidjson::UTF8<>>& json) {
  return json.HasMember("e") &&
         std::string_view(json["e"].GetString()) == "outboundAccountPosition";
}

bool IsOrderUpdate(const rapidjson::GenericValue<rapidjson::UTF8<>>& json) {
  return json.HasMember("e") && std::string(json["e"].GetString()) == "executionReport";
}

bool validate_raw_margin_entry(const JsonValue& doc) {
  if (!doc.HasMember("asset") || !doc.HasMember("free") || !doc.HasMember("locked") ||
      !doc.HasMember("netAsset")) {
    return false;
  } else {
    return true;
  }
}

}  // namespace

std::map<std::string, TransferType> BinanceParser::transfer_type_map_{
    {"MAIN_UMFUTURE", TransferRequestProto::WITHDRAW_SPOT},
    {"MAIN_CMFUTURE", TransferRequestProto::WITHDRAW_SPOT},
    {"MAIN_MARGIN", TransferRequestProto::WITHDRAW_SPOT},
    {"MAIN_MINING", TransferRequestProto::WITHDRAW_SPOT},
    {"C2C_MAIN", TransferRequestProto::DEPOSIT_SPOT},
    {"UMFUTURE_MAIN", TransferRequestProto::DEPOSIT_SPOT},
    {"CMFUTURE_MAIN", TransferRequestProto::DEPOSIT_SPOT},
    {"MARGIN_MAIN", TransferRequestProto::DEPOSIT_SPOT},
    {"MINING_MAIN", TransferRequestProto::DEPOSIT_SPOT},
    {"MAIN_PAY", TransferRequestProto::WITHDRAW_SPOT},
    {"PAY_MAIN", TransferRequestProto::DEPOSIT_SPOT}};

void BinanceParser::ParseRestSpotAccountBalance() {
  JsonDocument document;
  if (!PrepareJsonDocument(&document) || !document.IsObject() || IsBinanceErrMessage(document) ||
      !document.HasMember("balances") || !document["balances"].IsArray()) {
    LOG(ERROR) << "[ParseRestAccountBalance] Parsing Error. " << msg_->DebugString();
    result_.set_error_code(OrderErrorCode::INVALID_JSON);
    return;
  }
  // Linchuan: sometimes rest balance update slower than ws update about 10 sec,
  // makes balance "jump" back to wrong balance. solution: if ws balance is working,
  // use ws balance, if ws not update for 30 sec, use rest balance.
  if (GetCurrentTimestamp() - last_ws_balance_timestamp_ <= 30e9) {
    result_.set_type(MsgType::IGNORABLE);
    return;
  }

  auto account_balance = ParseBalance(document);

  if (is_readonly_order_system_ &&
      (!is_flexible_product_position_initialized_ || !is_locked_product_position_initialized_)) {
    result_.set_type(MsgType::IGNORABLE);
    return;
  }

  result_.set_type(MsgType::ACCOUNT_BALANCE);
  *result_.mutable_account_balance() = std::move(account_balance);
}

void BinanceParser::ParseRestEarnFlexibleProductPosition() {
  JsonDocument document;
  if (!PrepareJsonDocument(&document) || !document.IsObject() || !document.HasMember("rows") ||
      !document["rows"].IsArray()) {
    LOG(ERROR) << "[" << __func__ << "] Parsing Error. " << msg_->DebugString();
    result_.set_error_code(OrderErrorCode::INVALID_JSON);
    return;
  }

  const auto& array = document["rows"].GetArray();
  flexible_product_position_map_cache_.clear();

  for (const auto& data : array) {
    if (!data.HasMember("asset") || !data.HasMember("totalAmount") ||
        !data.HasMember("yesterdayRealTimeRewards") || !data.HasMember("cumulativeTotalRewards")) {
      continue;
    }
    auto& currency_balance = flexible_product_position_map_cache_[data["asset"].GetString()];

    auto avail = stod(data["totalAmount"].GetString());
    auto last_drawn_interest = stod(data["yesterdayRealTimeRewards"].GetString());
    auto cumulative_interest = stod(data["cumulativeTotalRewards"].GetString());

    currency_balance.set_currency_native(data["asset"].GetString());
    currency_balance.set_available(avail);
    currency_balance.set_total(avail);
    currency_balance.set_last_drawn_interest(last_drawn_interest);
    currency_balance.set_cumulative_interest(cumulative_interest);
  }

  is_flexible_product_position_initialized_ = true;

  if (!is_account_balance_initialized_ || !is_locked_product_position_initialized_) {
    result_.set_type(MsgType::IGNORABLE);
    return;
  }

  auto account_balance = GetAccountBalance();

  result_.set_type(MsgType::ACCOUNT_BALANCE);
  *result_.mutable_account_balance() = std::move(account_balance);
}

void BinanceParser::ParseRestEarnLockedProductPosition() {
  JsonDocument document;
  if (!PrepareJsonDocument(&document) || !document.IsObject() || !document.HasMember("rows") ||
      !document["rows"].IsArray()) {
    LOG(ERROR) << "[" << __func__ << "] Parsing Error. " << msg_->DebugString();
    result_.set_error_code(OrderErrorCode::INVALID_JSON);
    return;
  }

  const auto& array = document["rows"].GetArray();
  locked_product_position_map_cache_.clear();

  for (const auto& data : array) {
    if (!data.HasMember("asset") || !data.HasMember("amount")) {
      continue;
    }
    auto& currency_balance = locked_product_position_map_cache_[data["asset"].GetString()];

    auto hold = stod(data["amount"].GetString());

    currency_balance.set_currency_native(data["asset"].GetString());
    currency_balance.set_hold(hold);
    currency_balance.set_total(hold);
  }

  is_locked_product_position_initialized_ = true;

  if (!is_account_balance_initialized_ || !is_flexible_product_position_initialized_) {
    result_.set_type(MsgType::IGNORABLE);
    return;
  }

  auto account_balance = GetAccountBalance();

  result_.set_type(MsgType::ACCOUNT_BALANCE);
  *result_.mutable_account_balance() = std::move(account_balance);
}

AccountBalance BinanceParser::GetAccountBalance() {
  AccountBalance acc_balance;
  acc_balance.set_exchange("Binance");
  acc_balance.set_market_type("Spot");
  for (const auto& native_currency : symcache_->GetCurrenciesNativeUpper()) {
    auto* balance = acc_balance.add_each_balance();
    if (account_balance_map_cache_.count(native_currency) == 0) {
      balance->set_total(0);
      balance->set_hold(0);
      balance->set_available(0);
    } else {
      balance->MergeFrom(account_balance_map_cache_.at(native_currency));
    }

    MergeEarnPosition(native_currency, balance);

    auto* currency = symcache_->GetCurrencyByNative(native_currency);
    balance->set_currency(currency->currency());
    balance->set_currency_native(currency->native_currency());
    DLOG(INFO) << "BALANCE: " << currency->native_currency() << " " << balance->total() << " "
               << balance->available() << " " << balance->hold() << " "
               << balance->last_drawn_interest() << " " << balance->cumulative_interest();
  }

  return acc_balance;
}

void BinanceParser::MergeEarnPosition(
    const std::string& native_currency,
    CurrencyBalance* balance) {
  if (flexible_product_position_map_cache_.count(native_currency) != 0) {
    const auto& currency_balance = flexible_product_position_map_cache_[native_currency];

    balance->set_total(balance->total() + currency_balance.total());
    balance->set_available(balance->available() + currency_balance.available());
    balance->set_last_drawn_interest(currency_balance.last_drawn_interest());
    balance->set_cumulative_interest(currency_balance.cumulative_interest());
  }

  if (locked_product_position_map_cache_.count(native_currency) != 0) {
    const auto& currency_balance = locked_product_position_map_cache_[native_currency];

    balance->set_total(balance->total() + currency_balance.total());
    balance->set_hold(balance->hold() + currency_balance.hold());
  }
}

void BinanceParser::ParseRestMarginCrossedAccountBalance() {
  JsonDocument document;
  if (!PrepareJsonDocument(&document)) {
    LOG(ERROR) << "[ParseRestMarginCrossedAccountBalance] Parsing Error. " << msg_->DebugString();
    result_.set_error_code(OrderErrorCode::INVALID_JSON);
    return;
  }

  RawBalanceMap raw_balance_map;
  auto success = ParseRawMarginCrossedBalance(document, &raw_balance_map);
  if (!success) {
    return;
  }

  crossed_margin_balance_map_cache_ = std::move(raw_balance_map);
  is_crossed_margin_balance_initialized = true;
  if (!is_isolated_margin_balance_initialized ||
      (is_readonly_order_system_ &&
       (!is_flexible_product_position_initialized_ || !is_locked_product_position_initialized_))) {
    return;
  }

  AccountBalance acc_balance;
  MergeMarginAccountBalance(&acc_balance);
  result_.set_type(MsgType::ACCOUNT_BALANCE);
  *result_.mutable_account_balance() = std::move(acc_balance);
}

void BinanceParser::ParseRestMarginIsolatedAccountBalance() {
  JsonDocument document;
  if (!PrepareJsonDocument(&document)) {
    LOG(ERROR) << "[ParseRestMarginIsolatedAccountBalance] Parsing Error. " << msg_->DebugString();
    result_.set_error_code(OrderErrorCode::INVALID_JSON);
    return;
  }

  RawBalanceMap raw_balance_map;
  auto success = ParseRawMarginIsolatedBalance(document, &raw_balance_map);
  if (!success) {
    return;
  }

  isolated_margin_balance_map_cache_ = std::move(raw_balance_map);
  is_isolated_margin_balance_initialized = true;
  if (!is_crossed_margin_balance_initialized ||
      (is_readonly_order_system_ &&
       (!is_flexible_product_position_initialized_ || !is_locked_product_position_initialized_))) {
    return;
  }

  AccountBalance acc_balance;
  MergeMarginAccountBalance(&acc_balance);

  result_.set_type(MsgType::ACCOUNT_BALANCE);
  *result_.mutable_account_balance() = std::move(acc_balance);
}

bool BinanceParser::ParseRawMarginCrossedBalance(
    const JsonValue& doc,
    RawBalanceMap* raw_balance_map) {
  /*
  {
      "borrowEnabled": true,
      "marginLevel": "11.64405625",
      "totalAssetOfBtc": "6.82728457",
      "totalLiabilityOfBtc": "0.58633215",
      "totalNetAssetOfBtc": "6.24095242",
      "tradeEnabled": true,
      "transferEnabled": true,
      "userAssets": [
          {
              "asset": "BTC",
              "borrowed": "0.00000000",
              "free": "0.00499500",
              "interest": "0.00000000",
              "locked": "0.00000000",
              "netAsset": "0.00499500"
          },
          ...
      ]
  }
  */
  raw_balance_map->clear();
  if (doc.IsObject() && IsBinanceErrMessage(doc)) {
    LOG(ERROR) << "[ParseMarginCrossedBalance] " << JsonToString(doc);
    return false;
  }

  if (!doc.HasMember("userAssets") || !doc["userAssets"].IsArray()) {
    LOG(ERROR) << "[ParseMarginCrossedBalance] " << JsonToString(doc);
    return false;
  }

  for (const auto& data : doc["userAssets"].GetArray()) {
    if (!validate_raw_margin_entry(data)) {
      continue;
    }
    auto current_native_currency = data["asset"].GetString();
    CHECK(raw_balance_map->count(current_native_currency) == 0) << current_native_currency;
    CurrencyBalance balance;
    balance.set_currency_native(current_native_currency);
    balance.set_available(stod(data["free"].GetString()));
    balance.set_hold(stod(data["locked"].GetString()));
    balance.set_total(stod(data["netAsset"].GetString()));
    raw_balance_map->emplace(balance.currency_native(), balance);
  }

  return true;
}

bool BinanceParser::ParseRawMarginIsolatedBalance(
    const JsonValue& doc,
    RawBalanceMap* raw_balance_map) {
  /*
  {
   "assets":[
      {
        "baseAsset":
        {
          "asset": "BTC",
          "borrowEnabled": true,
          "borrowed": "0.00000000",
          "free": "0.00000000",
          "interest": "0.00000000",
          "locked": "0.00000000",
          "netAsset": "0.00000000",
          "netAssetOfBtc": "0.00000000",
          "repayEnabled": true,
          "totalAsset": "0.00000000"
        },
        "quoteAsset":
        {
          "asset": "USDT",
          "borrowEnabled": true,
          "borrowed": "0.00000000",
          "free": "0.00000000",
          "interest": "0.00000000",
          "locked": "0.00000000",
          "netAsset": "0.00000000",
          "netAssetOfBtc": "0.00000000",
          "repayEnabled": true,
          "totalAsset": "0.00000000"
        },
        "symbol": "BTCUSDT",
        "isolatedCreated": true,
        "enabled": true, // true-enabled, false-disabled
        "marginLevel": "0.00000000",
        "marginLevelStatus": "EXCESSIVE", // "EXCESSIVE", "NORMAL", "MARGIN_CALL",
"PRE_LIQUIDATION", "FORCE_LIQUIDATION" "marginRatio": "0.00000000", "indexPrice": "10000.00000000",
        "liquidatePrice": "1000.00000000",
        "liquidateRate": "1.00000000",
        "tradeEnabled": true
      }
    ],
    "totalAssetOfBtc": "0.00000000",
    "totalLiabilityOfBtc": "0.00000000",
    "totalNetAssetOfBtc": "0.00000000"
}
  */
  raw_balance_map->clear();
  if (doc.IsObject() && IsBinanceErrMessage(doc)) {
    LOG(ERROR) << "[ParseMarginIsolatedBalance] " << JsonToString(doc);
    return false;
  }

  if (!doc.HasMember("assets") || !doc["assets"].IsArray()) {
    LOG(ERROR) << "[ParseMarginIsolatedBalance] " << JsonToString(doc);
    return false;
  }

  for (const auto& data : doc["assets"].GetArray()) {
    if (!data.HasMember("baseAsset") || !data.HasMember("quoteAsset")) {
      continue;
    }
    if (!validate_raw_margin_entry(data["baseAsset"]) ||
        !validate_raw_margin_entry(data["quoteAsset"])) {
      continue;
    }
    ParseRawMarginIsolatedAssets(data["baseAsset"], raw_balance_map);
    ParseRawMarginIsolatedAssets(data["quoteAsset"], raw_balance_map);
  }

  return true;
}

bool BinanceParser::ParseRawMarginIsolatedAssets(
    const JsonValue& doc,
    RawBalanceMap* raw_balance_map) {
  auto current_native_currency = doc["asset"].GetString();
  auto& balance = (*raw_balance_map)[current_native_currency];
  balance.set_currency_native(current_native_currency);
  balance.set_available(stod(doc["free"].GetString()) + balance.available());
  balance.set_hold(stod(doc["locked"].GetString()) + balance.hold());
  balance.set_total(stod(doc["netAsset"].GetString()) + balance.total());
  return true;
}

void BinanceParser::MergeMarginAccountBalance(AccountBalance* acc_balance) {
  acc_balance->Clear();
  acc_balance->set_exchange("Binance");
  acc_balance->set_market_type("Margin");
  for (const auto& native_currency : symcache_->GetCurrenciesNativeUpper()) {
    auto* balance = acc_balance->add_each_balance();
    balance->set_total(0);
    balance->set_hold(0);
    balance->set_available(0);
    if (isolated_margin_balance_map_cache_.count(native_currency)) {
      auto& tmp = isolated_margin_balance_map_cache_.at(native_currency);
      balance->set_total(balance->total() + tmp.total());
      balance->set_hold(balance->hold() + tmp.hold());
      balance->set_available(balance->available() + tmp.available());
    }
    if (crossed_margin_balance_map_cache_.count(native_currency)) {
      auto& tmp = crossed_margin_balance_map_cache_.at(native_currency);
      balance->set_total(balance->total() + tmp.total());
      balance->set_hold(balance->hold() + tmp.hold());
      balance->set_available(balance->available() + tmp.available());
    }

    MergeEarnPosition(native_currency, balance);

    auto* currency = symcache_->GetCurrencyByNative(native_currency);
    balance->set_currency(currency->currency());
    balance->set_currency_native(currency->native_currency());
    DLOG(INFO) << "BALANCE: " << currency->native_currency() << " " << balance->total() << " "
               << balance->available() << " " << balance->hold() << " "
               << balance->last_drawn_interest() << " " << balance->cumulative_interest();
  }
}

AccountBalance BinanceParser::ParseBalance(const JsonValue& doc) {
  account_balance_map_cache_.clear();

  for (const auto& data : doc["balances"].GetArray()) {
    if (!data.HasMember("asset") || !data.HasMember("free") || !data.HasMember("locked")) {
      continue;
    }
    auto current_free = stod(data["free"].GetString());
    auto current_locked = stod(data["locked"].GetString());

    auto& balance = account_balance_map_cache_[data["asset"].GetString()];

    balance.set_currency_native(data["asset"].GetString());
    balance.set_available(current_free);
    balance.set_hold(current_locked);
    balance.set_total(current_free + current_locked);
  }
  is_account_balance_initialized_ = true;

  return GetAccountBalance();
}

bool BinanceParser::ParseRawBalanceFromWs(const JsonValue& doc, RawBalanceMap* raw_balance_map) {
  /*
  {
    "e":"outboundAccountInfo",
    "E":1594263002528,
    "m":2,
    "t":6,
    "b":0,
    "s":0,
    "T":true,
    "W":true,
    "D":true,
    "u":1594263002526,
    "B":[
        {
            "a":"BTC",
            "f":"1.35230710",
            "l":"0.00000000"
        },
        {
            "a":"LTC",
            "f":"0.00000000",
            "l":"0.00000000"
        },
        ...
    ]
  }

  From 2020-09-09
  {
    "e": "outboundAccountPosition", //Event type
    "E": 1564034571105,             //Event Time
    "u": 1564034571073,             //Time of last account update
    "B": [                          //Balances Array
      {
        "a": "ETH",                 //Asset
        "f": "10000.000000",        //Free
        "l": "0.000000"             //Locked
      }
    ]
  }
  */
  if (!(doc.HasMember("E"))) {
    LOG(ERROR) << JsonToString(doc);
    return false;
  }
  if (!(doc.HasMember("B") && doc["B"].IsArray())) {
    LOG(ERROR) << JsonToString(doc);
    return false;
  }

  for (const auto& currency_bal : doc["B"].GetArray()) {
    auto free = stod(currency_bal["f"].GetString());
    auto locked = stod(currency_bal["l"].GetString());
    CurrencyBalance balance;
    balance.set_currency_native(currency_bal["a"].GetString());
    balance.set_available(free);
    balance.set_hold(locked);
    balance.set_total(free + locked);
    raw_balance_map->emplace(balance.currency_native(), balance);
  }

  return true;
}

bool BinanceParser::ParseWsBalance(const JsonValue& doc, AccountBalance* acc_balance) {
  acc_balance->Clear();

  RawBalanceMap raw_balance_map;
  bool success = ParseRawBalanceFromWs(doc, &raw_balance_map);
  if (!success) {
    return false;
  }

  // update account_balance_map_cache_
  for (const auto& p : raw_balance_map) {
    account_balance_map_cache_[p.first] = p.second;
  }

  // update account balance, just publish delta change from ws update
  acc_balance->set_exchange("Binance");
  acc_balance->set_market_type("Spot");
  for (const auto& native_currency : symcache_->GetCurrenciesNativeUpper()) {
    if (raw_balance_map.count(native_currency) == 0) {
      continue;
    }
    auto* balance = acc_balance->add_each_balance();
    balance->MergeFrom(raw_balance_map.at(native_currency));

    MergeEarnPosition(native_currency, balance);

    auto* currency = symcache_->GetCurrencyByNative(native_currency);
    balance->set_currency(currency->currency());
    balance->set_currency_native(currency->native_currency());
    DLOG(INFO) << "BALANCE: " << currency->native_currency() << " " << balance->total() << " "
               << balance->available() << " " << balance->hold() << " "
               << balance->last_drawn_interest() << " " << balance->cumulative_interest();
  }

  return true;
}

ProductOrderElement BinanceParser::ParseRestOrderInfo(const JsonValue& data) {
  ProductOrderElement order;
  std::string native_symbol = data["symbol"].GetString();
  if (!symcache_->has_product_native(native_symbol)) {
    return order;
  }
  std::string exchange_id = fmt::format("{}{}", native_symbol, data["orderId"].GetString());
  if (GetCurrentTimestamp() / 1e9 - stoll(data["time"].GetString()) / 1e3 > 60 * 60) {
    auto* oc = GetOcm()->FindByExchangeOrderId(exchange_id);
    if (oc != nullptr && !GetOcm()->IsInWorkingOrders(oc)) {
      // TODO(xguo): adjust query url not to list all previous orders. Otherwise too chatty.
      // LOG(INFO) << "Ignore binance long time order: " << exchange_id;
      return order;
    }
  }

  order.set_market_type(MarketType::Spot);
  order.set_exchange_type(ExchangeType::Binance);
  order.set_api_version("v1");
  order.set_exchange_order_id(exchange_id);
  order.set_symbol(native_symbol);
  order.set_price(stod(data["price"].GetString()));
  order.set_qty(stod(data["origQty"].GetString()));
  order.set_proc_order_id(GetProcOrderId(data));
  order.set_side(GetOrderSide(data["side"].GetString()));
  order.set_order_type(GetOrderType(data["type"].GetString()));
  order.set_timestamp(GetCurrentTimestamp());
  SetOrderStatus(data["status"].GetString(), &order);

  const int64_t et = stoll(data["updateTime"].GetString()) * 1'000'000LL;
  rest_transaction_update_et_ = std::max(rest_transaction_update_et_, et);

  return order;
}

void BinanceParser::ParseRestOpenOrders() {
  is_query_historical_orders_ = false;
  ParseRestOrderInfoList();
}

void BinanceParser::ParseRestAllOrders() {
  is_query_historical_orders_ = true;
  ParseRestOrderInfoList();
}

void BinanceParser::ParseRestOrderInfoList() {
  JsonDocument data;
  if (!PrepareJsonDocument(&data)) {
    LOG(ERROR) << "[ParseRestOpenOrders] Parsing Error. " << msg_->DebugString();
    result_.set_error_code(OrderErrorCode::INVALID_JSON);
    return;
  }
  if (data.IsObject() && IsBinanceErrMessage(data)) {
    LOG(ERROR) << "[ParseRestOrderInfoList] " << JsonToString(data);
    return;
  }
  if (!data.IsArray()) {
    LOG(ERROR) << "Unexpected order list format!" << JsonToString(data);
    return;
  }

  auto product_order_bundle = result_.mutable_order_list();
  product_order_bundle->Clear();

  bool product_order_bundle_found = false;
  for (const auto& info : data.GetArray()) {
    auto new_order = ParseRestOrderInfo(info);

    // Ignore open orders from QueryHistoricalOrders
    // only trust open roders from QueryOpenOrders
    if (!(is_query_historical_orders_ && new_order.is_live()) && new_order.has_market_type()) {
      auto* order = product_order_bundle->add_each_order();
      order->MergeFrom(new_order);
      product_order_bundle_found = true;
    }
  }
  if (product_order_bundle_found) {
    result_.set_type(MsgType::ORDER_STATUS);
  }
}

ProductOrderBundle BinanceParser::ParseWsOrderInfoListV2(const JsonValue& data) {
  ProductOrderBundle product_order_bundle;

  auto* order = product_order_bundle.add_each_order();
  order->set_market_type(MarketType::Spot);
  order->set_exchange_type(ExchangeType::Binance);
  order->set_api_version("v1");
  order->set_exchange_order_id(data["s"].GetString() + std::string(data["i"].GetString()));
  order->set_symbol(data["s"].GetString());
  order->set_price(stod(data["p"].GetString()));
  order->set_qty(stod(data["q"].GetString()));
  if (std::string(data["C"].GetString()).empty()) {
    order->set_proc_order_id(stoll(data["c"].GetString()));
  } else {
    order->set_proc_order_id(stoll(data["C"].GetString()));
  }
  order->set_side(GetOrderSide(data["S"].GetString()));
  order->set_order_type(GetOrderType(data["o"].GetString()));
  SetOrderStatus(data["X"].GetString(), order);
  order->set_timestamp(GetCurrentTimestamp());

  const int64_t et = stoll(data["T"].GetString()) * 1000'000LL;
  wss_transaction_update_et_ = std::max(wss_transaction_update_et_, et);

  return product_order_bundle;
}

ProductFillBundle BinanceParser::ParseWsFillInfoListV2(const JsonValue& data) {
  ProductFillBundle product_fill_bundle;

  std::string status = data["x"].GetString();
  if (status == "TRADE" || status == "AGGTRADE") {
    auto* fill = product_fill_bundle.add_each_fill();
    fill->set_symbol(data["s"].GetString());
    fill->set_fill_timestamp(std::stoll(data["E"].GetString()) * 1000000LL);
    fill->set_exchange_order_id(data["s"].GetString() + std::string(data["i"].GetString()));
    if (std::string(data["C"].GetString()).empty()) {
      fill->set_proc_order_id(stoll(data["c"].GetString()));
    } else {
      fill->set_proc_order_id(stoll(data["C"].GetString()));
    }
    fill->set_price(stod(data["L"].GetString()));
    fill->set_qty(stod(data["l"].GetString()));
    fill->set_fill_id(data["t"].GetString());
    if (data["m"].GetBool()) {
      fill->set_fill_type(FillType::MAKER_FILL_TYPE);
    } else {
      fill->set_fill_type(FillType::TAKER_FILL_TYPE);
    }
    if (data.HasMember("N") && !data["N"].IsNull()) {
      const std::string fee_currency = data["N"].GetString();
      if (!fee_currency.empty() && fee_currency != "BNB") {
        fill->set_fee_currency(fee_currency);
        if (data.HasMember("n") && !data["n"].IsNull()) {
          const std::string fee = data["n"].GetString();
          if (!fee.empty()) {
            fill->set_fee(stod(fee));
          }
        }
      }
    }
  }

  const int64_t et = stoll(data["T"].GetString()) * 1000'000LL;
  wss_transaction_update_et_ = std::max(wss_transaction_update_et_, et);

  return product_fill_bundle;
}

void BinanceParser::ParseRestFillInfoList() {
  JsonDocument data;
  if (!PrepareJsonDocument(&data)) {
    LOG(ERROR) << "[ParseRestFillOrders] Parsing Error. " << msg_->DebugString();
    result_.set_error_code(OrderErrorCode::INVALID_JSON);
    return;
  }
  if (data.IsObject() && IsBinanceErrMessage(data)) {
    LOG(ERROR) << "[ParseRestFillInfoList] " << JsonToString(data);
    return;
  }
  if (!(data.IsArray() && data.Size() > 0)) {
    return;
  }

  auto product_fill_bundle = result_.mutable_fill_list();
  product_fill_bundle->Clear();

  bool product_fill_bundle_found = false;
  for (const auto& info : data.GetArray()) {
    auto* fill = product_fill_bundle->add_each_fill();
    fill->set_symbol(info["symbol"].GetString());
    fill->set_fill_timestamp(std::stoll(info["time"].GetString()) * 1000000LL);
    fill->set_exchange_order_id(
        info["symbol"].GetString() + std::string(info["orderId"].GetString()));
    fill->set_price(stod(info["price"].GetString()));
    fill->set_qty(stod(info["qty"].GetString()));
    fill->set_fill_id(info["id"].GetString());
    if (info["isMaker"].GetBool()) {
      fill->set_fill_type(FillType::MAKER_FILL_TYPE);
    } else {
      fill->set_fill_type(FillType::TAKER_FILL_TYPE);
    }
    if (info.HasMember("commissionAsset") && !info["commissionAsset"].IsNull()) {
      const std::string fee_currency = info["commissionAsset"].GetString();
      if (!fee_currency.empty() && fee_currency != "BNB") {
        fill->set_fee_currency(fee_currency);
        if (info.HasMember("commission") && !info["commission"].IsNull()) {
          const std::string fee = info["commission"].GetString();
          if (!fee.empty()) {
            fill->set_fee(stod(fee));
          }
        }
      }
    }
    product_fill_bundle_found = true;

    const int64_t et = stoll(info["time"].GetString()) * 1'000'000LL;
    rest_transaction_update_et_ = std::max(rest_transaction_update_et_, et);
  }
  if (product_fill_bundle_found) {
    result_.set_type(MsgType::ORDER_FILL);
  }
}

bool BinanceParser::IsValidBinanceResponse(const std::shared_ptr<AsioHttpContext>& http_context) {
  /*
  http_response {
    timestamp: 1679641708204068428
    status_code: 503
    headers {
      key: "Connection"
      value: "keep-alive"
    }
    headers {
      key: "Content-Length"
      value: "963"
    }
    headers {
      key: "Content-Type"
      value: "text/html"
    }
    headers {
      key: "Date"
      value: "Fri, 24 Mar 2023 07:08:28 GMT"
    }
    headers {
      key: "Server"
      value: "CloudFront"
    }
    headers {
      key: "Via"
      value: "1.1 6259584a0d6b38c5f298323a776f2b74.cloudfront.net (CloudFront)"
    }
    headers {
      key: "X-Amz-Cf-Id"
      value: "EYnZG6f6l5NWBG9FgxFhbIbLqESLEO7CLoW-0AyKMcHTE0Xn0kjf7A=="
    }
    headers {
      key: "X-Amz-Cf-Pop"
      value: "NRT20-C1"
    }
    headers {
      key: "X-Cache"
      value: "LimitExceeded from cloudfront"
    }
    body: "<!DOCTYPE HTML PUBLIC \"-//W3C//DTD HTML 4.01 Transitional//EN\"
  \"http://www.w3.org/TR/html4/loose.dtd\">\n<HTML><HEAD><META HTTP-EQUIV=\"Content-Type\"
  CONTENT=\"text/html; charset=iso-8859-1\">\n<TITLE>ERROR: The request could not be
  satisfied</TITLE>\n</HEAD><BODY>\n<H1>503 ERROR</H1>\n<H2>The request could not be
  satisfied.</H2>\n<HR noshade size=\"1px\">\nThis distribution exceeded the limit on requests per
  second.\nWe can\'t connect to the server for this app or website at this time. There might be too
  much traffic or a configuration error. Try again later, or contact the app or website owner.\n<BR
  clear=\"all\">\nIf you provide content to customers through CloudFront, you can find steps to
  troubleshoot and help prevent this error by reviewing the CloudFront documentation.\n<BR
  clear=\"all\">\n<HR noshade size=\"1px\">\n<PRE>\nGenerated by cloudfront (CloudFront)\nRequest
  ID:
  EYnZG6f6l5NWBG9FgxFhbIbLqESLEO7CLoW-0AyKMcHTE0Xn0kjf7A==\n</PRE>\n<ADDRESS>\n</ADDRESS>\n</BODY></HTML>"
  }
  */

  // special handle 503 code with cloudfront
  if (http_context->res.result_int() == 503) {
    std::string body(http_context->res.body().data(), http_context->res.body().length());

    if (body.find("Generated by cloudfront") != std::string::npos ||
        body.find(R"("code":-1001,)") != std::string::npos ||
        body.find("Service Unavailable.") != std::string::npos) {
      return true;
    }
  }

  return IsValidResponse(http_context);
}

void BinanceParser::ParseSubmitResponse() {
  const auto order_context = msg_->oc;
  const auto& http_context = msg_->http_context;
  if (!IsValidBinanceResponse(http_context)) {
    return;
  }

  result_.set_type(MsgType::SUBMIT_RESPONSE);
  auto resp = result_.mutable_submit_response();
  resp->Clear();
  resp->set_symbol(order_context->order_spec().product);
  resp->set_proc_order_id(order_context->proc_order_id());
  resp->set_http_status(http_context->res.result_int());
  resp->set_timestamp(GetCurrentTimestamp());

  if (http_context->status == AsioHttpContextStatus::kError) {
    LOG(ERROR) << "[BinanceSubmitOrder] AsioHttpContextStatus Error";
    resp->set_error_code(coin::proto::OrderErrorCode::ORDER_NETWORK_ERROR);
    resp->set_success(false);
    return;
  }

  const auto http_code = http_context->res.result_int();
  if (http_code == 429 || http_code == 418) {
    LOG(ERROR) << "[BinanceSubmitOrder] API_LIMIT_ERROR";
    resp->set_error_code(coin::proto::OrderErrorCode::API_LIMIT_ERROR);
    resp->set_exchange_error_msg(http_context->res.body().data());
    resp->set_success(false);
    return;
  }

  JsonDocument json;
  if (!PrepareJsonDocument(&json)) {
    LOG(ERROR) << "[BinanceSubmitOrder] Parse Error\n" << http_context->res;
    resp->set_error_code(coin::proto::OrderErrorCode::ORDER_API_ERROR);
    resp->set_success(false);
    return;
  }

  if (json.HasMember("code") && json.HasMember("msg")) {
    auto error_code = GetOrderErrorCode(json["code"], json["msg"]);
    resp->set_error_code(error_code);
    resp->set_exchange_error_code(GetExchangeOrderErrorCode(json));
    resp->set_exchange_error_msg(GetExchangeOrderErrorMsg(json));
    if (error_code != coin::proto::OrderErrorCode::ORDER_NO_ERROR) {
      LOG(ERROR) << "[BinanceSubmitOrder] Error Response: \n" << http_context->res;
      resp->set_success(false);
    }
    resp->set_success(false);
    return;
  }

  resp->set_success(true);
  resp->set_exchange_order_id(
      json["symbol"].GetString() + std::string(json["orderId"].GetString()));

  const int64_t et = stoll(json["transactTime"].GetString()) * 1'000'000LL;
  rest_transaction_update_et_ = std::max(rest_transaction_update_et_, et);
}

std::string BinanceParser::ParseHeader(const std::shared_ptr<AsioHttpContext>& http_context) {
  const auto& header = http_context->res.base();
  nlohmann::json json = {
      {"x-mbx-order-count-1d", -1},
      {"x-mbx-order-count-10s", -1},
      {"x-mbx-order-count-1m", -1},
      {"x-mbx-used-weight-1m", -1},
      {"retry-after", -1},
      {"bind_ip", http_context->bind_ip}};

  if (header.count("x-mbx-used-weight-1m") > 0) {
    json["x-mbx-used-weight-1m"] = std::stoi(header.at("x-mbx-used-weight-1m").data());
  }

  if (header.count("x-mbx-order-count-10s") > 0) {
    json["x-mbx-order-count-10s"] = std::stoi(header.at("x-mbx-order-count-10s").data());
  }

  if (header.count("x-mbx-order-count-1d") > 0) {
    json["x-mbx-order-count-1d"] = std::stoi(header.at("x-mbx-order-count-1d").data());
  }

  if (header.count("retry-after") > 0) {
    json["retry-after"] = std::stoi(header.at("retry-after").data());
  } else if (http_context->res.result_int() == 403) {
    LOG(ERROR)
        << "WAF limit hit! block request for 6 minutes prevent WAF block time extend longer.";
    json["retry-after"] = 6 * 60;
  }
  return json.dump();
}

void BinanceParser::ParseCancelResponse() {
  const auto order_context = msg_->oc;
  const auto& http_context = msg_->http_context;
  if (!IsValidBinanceResponse(http_context)) {
    return;
  }

  result_.set_type(MsgType::CANCEL_RESPONSE);
  auto resp = result_.mutable_cancel_response();
  resp->Clear();
  resp->set_symbol(order_context->order_spec().product);
  resp->set_proc_order_id(order_context->proc_order_id());
  resp->set_http_status(http_context->res.result_int());
  resp->set_exchange_order_id(order_context->exchange_order_id());
  resp->set_timestamp(msg_->timestamp);

  if (http_context->status == AsioHttpContextStatus::kError) {
    LOG(ERROR) << "[BinanceCancelOrder] AsioHttpContextStatus Error";
    resp->set_error_code(coin::proto::OrderErrorCode::ORDER_NETWORK_ERROR);
    resp->set_success(false);
    return;
  }

  const auto http_code = http_context->res.result_int();
  if (http_code == 429 || http_code == 418) {
    LOG(ERROR) << "[BinanceCancelOrder] API_LIMIT_ERROR";
    resp->set_error_code(coin::proto::OrderErrorCode::API_LIMIT_ERROR);
    resp->set_exchange_error_msg(http_context->res.body().data());
    resp->set_success(false);
    return;
  }

  JsonDocument json;
  if (!PrepareJsonDocument(&json)) {
    LOG(ERROR) << "[BinanceCancelOrder] Parse Error\n" << http_context->res;
    resp->set_error_code(coin::proto::OrderErrorCode::ORDER_API_ERROR);
    resp->set_success(false);
    return;
  }
  if (json.HasMember("symbol") && json.HasMember("orderId")) {
    resp->set_exchange_order_id(
        json["symbol"].GetString() + std::string(json["orderId"].GetString()));
  }

  if (json.HasMember("code") && json.HasMember("msg")) {
    auto error_code = GetOrderErrorCode(json["code"], json["msg"]);
    resp->set_error_code(error_code);
    resp->set_exchange_error_code(GetExchangeOrderErrorCode(json));
    resp->set_exchange_error_msg(GetExchangeOrderErrorMsg(json));
    if (error_code != coin::proto::OrderErrorCode::ORDER_NO_ERROR) {
      LOG(ERROR) << "[BinanceCancelOrder] Error Response" << http_context->res;
      resp->set_success(false);
      return;
    }
  }

  resp->set_success(true);
  resp->set_cancel_confirmed(true);

  const int64_t et = stoll(json["transactTime"].GetString()) * 1'000'000LL;
  rest_transaction_update_et_ = std::max(rest_transaction_update_et_, et);
}

void BinanceParser::ParseRestDeposit() {
  /*
  [
    {
      "amount":"0.00999800",
      "coin":"PAXG",
      "network":"ETH",
      "status":1,
      "address":"0x788cabe9236ce061e5a892e1a59395a81fc8d62c",
      "addressTag":"",
      "txId":"0xaad4654a3234aa6118af9b4b335f5ae81c360b2394721c019b5d1e75328b09f3",
      "insertTime":1599621997000,
      "transferType":0,
      "unlockConfirm":"12/12",
      "confirmTimes":"12/12"
    }
  ]
  */
  JsonDocument document;
  if (!PrepareJsonDocument(&document)) {
    LOG(ERROR) << "[ParseRestDeposit] Parsing error\n" << msg_->DebugString();
    result_.set_error_code(OrderErrorCode::INVALID_JSON);
    return;
  }

  if (document.IsObject() && IsBinanceErrMessage(document)) {
    LOG(ERROR) << "[ParseRestDeposit] Error message\n" << JsonToString(document);
    result_.set_error_code(OrderErrorCode::INVALID_MSG_TYPE);
    return;
  }

  if (!document.IsArray()) {
    LOG(ERROR) << "[ParseRestDeposit] Invalid data format\n" << JsonToString(document);
    result_.set_error_code(OrderErrorCode::INVALID_MSG_TYPE);
    return;
  }

  auto currency_transfer_bundle = result_.mutable_currency_transfer_bundle();
  currency_transfer_bundle->Clear();
  bool currency_transfer_bundle_found = false;
  for (const auto& data : document.GetArray()) {
    const static std::vector<std::string> fields =
        {"coin", "txId", "status", "amount", "address", "insertTime"};
    if (!JsonFieldCheck(data, fields)) {
      LOG(ERROR) << "[ParseRestDeposit] Invalid Withdraw format" << JsonToString(data);
      continue;
    }

    std::string transfer_id = data["txId"].GetString();
    // Initialize status as empty
    if (transfer_deposit_map_.count(transfer_id) <= 0) {
      transfer_deposit_map_[transfer_id] = "";
    }
    std::string status = data["status"].GetString();
    // Skip it if not concerned
    if ((status != "0") && (status != "1") && (status != "6")) {
      continue;
    }
    // Skip it if duplicate
    if (!transfer_first_record_) {
      if (status == transfer_deposit_map_[transfer_id]) {
        continue;
      }
    }
    transfer_deposit_map_[transfer_id] = status;
    int64_t create_timestamp = stoll(data["insertTime"].GetString()) * 1e6;
    auto currency = BinanceCurrency::FromNativeStrImpl(data["coin"].GetString(), create_timestamp);
    auto* currency_transfer = currency_transfer_bundle->add_each_transfer();
    currency_transfer->set_currency(currency->currency());
    currency_transfer->set_type(TransferRequestProto::DEPOSIT);
    if (status == "0") {
      currency_transfer->set_status(TransferRequestProto::TRANSFER_SUBMITTED);
    } else if ((status == "6") || (status == "1")) {
      currency_transfer->set_status(TransferRequestProto::TRANSFER_CONFIRMED);
    }
    currency_transfer->set_external_transfer_id(transfer_id);
    currency_transfer->set_qty(stod(data["amount"].GetString()));
    currency_transfer->set_to_address(data["address"].GetString());
    currency_transfer->set_create_timestamp(create_timestamp);
    currency_transfer_bundle_found = true;
  }

  if (!transfer_first_record_) {
    if (!is_transfer_deposit_initialized_) {
      is_transfer_deposit_initialized_ = true;
      LOG(INFO) << "[ParseRestDeposit] Deposit ignored: " << transfer_deposit_map_.size();
      return;
    }
  }

  if (currency_transfer_bundle_found) {
    LOG(INFO) << "[ParseRestDeposit] Deposit parsed: "
              << currency_transfer_bundle->each_transfer_size();
    result_.set_type(MsgType::TRANSFER);
  }
  return;
}

void BinanceParser::ParseRestWithdrawal() {
  /*
  [
    {
      "address": "0x94df8b352de7f46f64b01d3666bf6e936e44ce60",
      "amount": "8.91000000",
      "applyTime": "2019-10-12 11:12:02",
      "coin": "USDT",
      "id": "b6ae22b3aa844210a7041aee7589627c",
      "withdrawOrderId": "WITHDRAWtest123",
      "network": "ETH",
      "transferType": 0,
      "status": 6,
      "transactionFee": "0.004",
      "confirmNo":3,
      "info":"The address is not valid. Please confirm with the recipient",
      "txId": "0xb5ef8c13b968a406cc62a93a8bd80f9e9a906ef1b3fcf20a2e48573c17659268"
    }
  ]
  */
  JsonDocument document;
  if (!PrepareJsonDocument(&document)) {
    LOG(ERROR) << "[ParseRestWithdrawal] Parsing error\n" << msg_->DebugString();
    result_.set_error_code(OrderErrorCode::INVALID_JSON);
    return;
  }

  if (document.IsObject() && IsBinanceErrMessage(document)) {
    LOG(ERROR) << "[ParseRestWithdrawal] Error message\n" << JsonToString(document);
    result_.set_error_code(OrderErrorCode::INVALID_MSG_TYPE);
    return;
  }

  if (!document.IsArray()) {
    LOG(ERROR) << "[ParseRestWithdrawal] Invalid data format\n" << JsonToString(document);
    result_.set_error_code(OrderErrorCode::INVALID_MSG_TYPE);
    return;
  }

  auto currency_transfer_bundle = result_.mutable_currency_transfer_bundle();
  currency_transfer_bundle->Clear();
  bool currency_transfer_bundle_found = false;
  for (const auto& data : document.GetArray()) {
    const static std::vector<std::string> fields =
        {"coin", "amount", "txId", "status", "address", "applyTime"};
    if (!JsonFieldCheck(data, fields)) {
      LOG(ERROR) << "[ParseRestWithdraw] Invalid Withdraw format" << JsonToString(data);
      continue;
    }

    std::string transfer_id = data["txId"].GetString();
    // Initialize status as empty
    if (transfer_withdrawal_map_.count(transfer_id) <= 0) {
      transfer_withdrawal_map_[transfer_id] = "";
    }
    std::string status = data["status"].GetString();
    // Skip it if not concerned
    if ((status != "1") && (status != "2") && (status != "3") && (status != "4") &&
        (status != "5") && (status != "6")) {
      continue;
    }
    // Skip it if duplicate
    if (!transfer_first_record_) {
      if (status == transfer_withdrawal_map_[transfer_id]) {
        continue;
      }
    }
    transfer_withdrawal_map_[transfer_id] = status;

    int64_t create_timestamp = GetTransferTimestamp(data["applyTime"].GetString());
    auto currency = BinanceCurrency::FromNativeStrImpl(data["coin"].GetString(), create_timestamp);
    auto* currency_transfer = currency_transfer_bundle->add_each_transfer();
    currency_transfer->set_currency(currency->currency());
    currency_transfer->set_type(TransferRequestProto::WITHDRAW);
    if (status == "1") {
      currency_transfer->set_status(TransferRequestProto::CANCEL_CONFIRMED);
    } else if ((status == "3") || (status == "5")) {
      currency_transfer->set_status(TransferRequestProto::TRANSFER_REJECTED);
    } else if ((status == "2") || (status == "4")) {
      currency_transfer->set_status(TransferRequestProto::TRANSFER_SUBMITTED);
    } else if (status == "6") {
      currency_transfer->set_status(TransferRequestProto::TRANSFER_CONFIRMED);
    }
    currency_transfer->set_external_transfer_id(transfer_id);
    currency_transfer->set_qty(stod(data["amount"].GetString()));
    currency_transfer->set_to_address(data["address"].GetString());
    currency_transfer->set_create_timestamp(create_timestamp);
    currency_transfer_bundle_found = true;
  }

  if (!transfer_first_record_) {
    if (!is_transfer_withdrawal_initialized_) {
      is_transfer_withdrawal_initialized_ = true;
      LOG(INFO) << "[ParseRestWithdrawal] Withdrawal ignored: " << transfer_withdrawal_map_.size();
      return;
    }
  }

  if (currency_transfer_bundle_found) {
    LOG(INFO) << "[ParseRestWithdrawal] Withdrawal parsed: "
              << currency_transfer_bundle->each_transfer_size();
    result_.set_type(MsgType::TRANSFER);
  }
}

void BinanceParser::CheckAndForceResetWssConnection() {
  if (wss_transaction_update_et_ == 0 ||
      wss_transaction_update_et_ + 300'000'000'000LL /* 5 mins */ > rest_transaction_update_et_) {
    return;
  }
  mb_->PublishEvent("force_reset_wss_connection", &wss_transaction_update_et_ /* dummy */);
}

void BinanceParser::ParseWsMessage() {
  JsonDocument document;
  if (!PrepareJsonDocument(&document)) {
    LOG(ERROR) << "[ParseWsMessage] Parsing Error. " << msg_->DebugString();
    result_.set_error_code(OrderErrorCode::INVALID_JSON);
    return;
  }

  bool update_ws_last_recv_ts = false;
  if (IsAccountUpdate(document)) {
    // make first rest balance result before ws balance result,
    // so SetAllZeroBalance won't take effect on partial symbol
    if (!is_account_balance_initialized_ ||
        (is_readonly_order_system_ && (!is_flexible_product_position_initialized_ ||
                                       !is_locked_product_position_initialized_))) {
      result_.set_type(MsgType::IGNORABLE);
    } else {
      if (ParseWsBalance(document, result_.mutable_account_balance())) {
        result_.set_type(MsgType::ACCOUNT_BALANCE);
        last_ws_balance_timestamp_ = GetCurrentTimestamp();
      } else {
        result_.set_error_code(OrderErrorCode::INVALID_JSON);
      }
    }
    update_ws_last_recv_ts = true;
  } else if (IsOrderUpdate(document)) {
    result_.fill_bundle_ = ParseWsFillInfoListV2(document);
    result_.order_bundle_ = ParseWsOrderInfoListV2(document);
    result_.set_type(MsgType::ORDER_STATUS);
    result_.set_type(MsgType::ORDER_FILL);
    update_ws_last_recv_ts = true;
  } else if (IsSuccessDataRequestResp(document)) {
    result_.set_type(MsgType::IGNORABLE);
    update_ws_last_recv_ts = true;
  }

  if (update_ws_last_recv_ts) {
    int64_t timestamp = msg_->timestamp;
    mb_->PublishEvent("upd_ws_last_recv_msg_ts", &timestamp);
  }
}

void BinanceParser::ParseListenKey() {
  JsonDocument document;
  if (!PrepareJsonDocument(&document)) {
    LOG(ERROR) << "[ParseListenKey] Parsing Error. " << msg_->DebugString();
    result_.set_error_code(OrderErrorCode::INVALID_JSON);
    return;
  }
  // {"listenKey":"fmt5gmPtRPVO6QOf1bP2euju10VoZ0xCvXmh6f3LrvGEyeuh3tgOh3liN06O"}
  LOG(INFO) << "binance/spot: " << JsonToString(document);
  if (document.HasMember("listenKey")) {
    std::string listen_key(document["listenKey"].GetString());
    if (mb_) {
      mb_->PublishEvent("binance/listen_key", &listen_key);
    }
  } else {
    LOG(INFO) << "Failed to fetch listen key" << JsonToString(document);
  }
}

void BinanceParser::ParseTransfer(bool is_deposit_subaccount) {
  /*
    {
    "result": [
      {
        "tranId": 116280229347,
        "fromEmail": "coin-binance-80@prestolabs.io",
        "toEmail": "prestoinvt@gmail.com",
        "asset": "USDT",
        "amount": "396688.7490991",
        "createTimeStamp": 1662626664000,
        "fromAccountType": "USDT_FUTURE",
        "toAccountType": "SPOT",
        "status": "SUCCESS"
      },
      {
        "tranId": 116280223735,
        "fromEmail": "coin-binance-79@prestolabs.io",
        "toEmail": "prestoinvt@gmail.com",
        "asset": "USDT",
        "amount": "396806.10167825",
        "createTimeStamp": 1662626657000,
        "fromAccountType": "USDT_FUTURE",
        "toAccountType": "SPOT",
        "status": "SUCCESS"
      }
    ],
      "totalCount": 97
    }
  */

  JsonDocument document;
  if (!PrepareJsonDocument(&document)) {
    LOG(ERROR) << "[ParseTransfer] Parsing Error. " << msg_->DebugString();
    result_.set_error_code(OrderErrorCode::INVALID_JSON);
    return;
  }

  if (document.IsObject() && IsBinanceErrMessage(document)) {
    LOG(ERROR) << "[ParseTransfer] Error message\n" << JsonToString(document);
    result_.set_error_code(OrderErrorCode::INVALID_MSG_TYPE);
    return;
  }

  if (!document["result"].IsArray()) {
    LOG(ERROR) << "[ParseTransfer] Invalid data format\n" << JsonToString(document);
    result_.set_error_code(OrderErrorCode::INVALID_MSG_TYPE);
    return;
  }

  auto currency_transfer_bundle = result_.mutable_currency_transfer_bundle();
  currency_transfer_bundle->Clear();
  bool currency_transfer_bundle_found = false;
  for (const auto& data : document["result"].GetArray()) {
    static std::vector<std::string> fields =
        {"asset", "amount", "createTimeStamp", "status", "tranId"};
    if (!JsonFieldCheck(data, fields)) {
      LOG(ERROR) << "[ParseRestTransfers] Invalid transfer format" << JsonToString(data);
      continue;
    }

    std::string status = data["status"].GetString();
    if (!transfer_first_record_) {
      if (transfer_account_map_[data["tranId"].GetString()] == status) {
        continue;
      }
    }

    transfer_account_map_.insert_or_assign(data["tranId"].GetString(), status);
    auto proto_status = TransferRequestProto::TRANSFER_SUBMITTED;
    if (status == "SUCCESS") {
      proto_status = TransferRequestProto::TRANSFER_CONFIRMED;
    } else if (status == "INIT") {
      proto_status = TransferRequestProto::TRANSFER_SUBMITTED;
    } else if (status == "FAILURE") {
      proto_status = TransferRequestProto::TRANSFER_REJECTED;
    } else if (status == "PROCESS") {
      proto_status = TransferRequestProto::TRANSFER_ACCEPTED;
    } else {
      LOG(ERROR) << "[ParseRestTransfers] Invalid transfer status" << JsonToString(data);
      continue;
    }

    int64_t create_timestamp = stoll(data["createTimeStamp"].GetString()) * 1e6;
    auto currency = BinanceCurrency::FromNativeStrImpl(data["asset"].GetString(), create_timestamp);
    auto* currency_transfer = currency_transfer_bundle->add_each_transfer();
    if (is_deposit_subaccount) {
      currency_transfer->set_type(TransferRequestProto::WITHDRAW_SUBACCOUNT);
    } else {
      currency_transfer->set_type(TransferRequestProto::DEPOSIT_SUBACCOUNT);
    }
    currency_transfer->set_currency(currency->currency());
    currency_transfer->set_qty(stod(data["amount"].GetString()));
    currency_transfer->set_create_timestamp(create_timestamp);
    currency_transfer->set_external_transfer_id(data["tranId"].GetString());
    currency_transfer->set_status(proto_status);

    if (data.HasMember("fromEmail")) {
      currency_transfer->set_from_address(data["fromEmail"].GetString());
    }
    if (data.HasMember("toEmail")) {
      currency_transfer->set_to_address(data["toEmail"].GetString());
    }

    currency_transfer_bundle_found = true;
  }

  if (!transfer_first_record_) {
    if (is_deposit_subaccount) {
      if (!is_transfer_deposit_subaccount_initialized_) {
        is_transfer_deposit_subaccount_initialized_ = true;
        LOG(INFO) << "[ParseRestTransfer] Transfer deposit subaccount ignored: "
                  << transfer_withdrawal_map_.size();
        return;
      }
    } else {
      if (!is_transfer_withdrawal_subaccount_initialized_) {
        is_transfer_withdrawal_subaccount_initialized_ = true;
        LOG(INFO) << "[ParseRestTransfer] Transfer withdrawal subaccount ignored: "
                  << transfer_withdrawal_map_.size();
        return;
      }
    }
  }

  if (currency_transfer_bundle_found) {
    LOG(INFO) << "[ParseTransfer] Transfer parsed: "
              << currency_transfer_bundle->each_transfer_size();
    result_.set_type(MsgType::TRANSFER);
  }
}

auto BinanceParser::ParseRawMessage(const RawMessage& msg) -> const ParseResult& {
  result_.Clear();
  msg_ = &msg;
  sub_topic_id_ = static_cast<OrderSubTopicId>(msg_->type);

  if (msg.type == OrderSubTopicId::WS_MESSAGE) {
    ParseWsMessage();
    return result_;
  }

  // make ParseHeader return some result;
  std::string s = ParseHeader(msg.http_context);
  if (mb_) {
    mb_->PublishEvent("binance/req_rate_limit", reinterpret_cast<void*>(&s));
  }

  switch (msg.http_context->query_type) {
    case OrderSubTopicId::LISTEN_KEY:
      ParseListenKey();
      break;
    case OrderSubTopicId::SUBMIT_RESPONSE:
      ParseSubmitResponse();
      break;
    case OrderSubTopicId::CANCEL_RESPONSE:
      ParseCancelResponse();
      break;
    case OrderSubTopicId::MULTI_CANCEL_RESPONSE:
      // pass
      break;
    case OrderSubTopicId::REST_ALL_ORDERS:
      ParseRestAllOrders();
      break;
    case OrderSubTopicId::REST_OPEN_ORDERS:
      ParseRestOpenOrders();
      break;
    case OrderSubTopicId::REST_FILL_ORDERS:
      ParseRestFillInfoList();
      break;
    case OrderSubTopicId::REST_ACCOUNT_BALANCE:
      ParseRestSpotAccountBalance();
      break;
    case BinanceOrderSubTopicId::REST_FLEXIBLE_PRODUCT_POSITION:
      ParseRestEarnFlexibleProductPosition();
      break;
    case BinanceOrderSubTopicId::REST_LOCKED_PRODUCT_POSITION:
      ParseRestEarnLockedProductPosition();
      break;
    case OrderSubTopicId::REST_ACCOUNT_CROSS_BALANCE:
      ParseRestMarginCrossedAccountBalance();
      break;
    case OrderSubTopicId::REST_ACCOUNT_NON_CROSS_BALANCE:
      ParseRestMarginIsolatedAccountBalance();
      break;
    case BinanceOrderSubTopicId::REST_DEPOSIT:
      ParseRestDeposit();
      break;
    case BinanceOrderSubTopicId::REST_WITHDRAWAL:
      ParseRestWithdrawal();
      break;
    case BinanceOrderSubTopicId::DEPOSIT_SUBACCOUNT:
      ParseTransfer(true);
      break;
    case BinanceOrderSubTopicId::WITHDRAW_SUBACCOUNT:
      ParseTransfer(false);
      break;
    default: {
      LOG(ERROR) << "Parsing Error. " << msg_->DebugString();
      result_.set_error_code(OrderErrorCode::ORDER_UNKNOWN_ERROR);
      break;
    }
  }

  CheckAndForceResetWssConnection();

  return result_;
}

}  // namespace coin2::exchange::binance::order_v1
